package skewreduce.dfof;

import java.io.IOException;
import java.nio.ByteOrder;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import skewreduce.framework.WritableInputFormat;
import skewreduce.framework.WritableOutputFormat;
import skewreduce.lib.DarkMatter;
import skewreduce.lib.GasParticle;
import skewreduce.lib.IPhysicsParticle;
import skewreduce.lib.PhysicsParticle;
import skewreduce.lib.Point3D;
import skewreduce.lib.StarParticle;



public class RewriteData extends Configured implements Tool {
	
	public static abstract class ParticleInputFormat<V extends Writable>
	extends WritableInputFormat<LongWritable,V> {
		protected LongWritable key = new LongWritable();
		
        public LongWritable createKey() { return key; }
        public int getKeySize() { return 8; }
        public ByteOrder getByteOrder() { return ByteOrder.LITTLE_ENDIAN; }		
	}
	
    public static final class GasInputFormat
    extends ParticleInputFormat<GasParticle> {
    	public GasParticle createValue() { return new GasParticle(); }
        public int getValueSize() { return 48; }
    }

    public static final class DarkInputFormat
    extends ParticleInputFormat<DarkMatter> {
        public DarkMatter createValue() { return new DarkMatter(); }
        public int getValueSize() { return 36; }
    }

    public static final class StarInputFormat
    extends ParticleInputFormat<StarParticle> {
        public StarParticle createValue() { return new StarParticle(); }
        public int getValueSize() { return 44; }
    }
    
    enum DataType {
    	GAS,
    	DARK,
    	STAR;
    	
    	public Class<? extends InputFormat<?,?>> getInputFormatClass() {
    		switch ( this ) {
    			case GAS: return GasInputFormat.class;
    			case DARK: return DarkInputFormat.class;
    			case STAR: return StarInputFormat.class;
    		}
    		throw new IllegalStateException();
    	}
    }
    
    public static class RewritePointMapper<V extends IPhysicsParticle> extends Mapper<LongWritable,V,LongWritable,Point3D> {
    	private Point3D outBuf = new Point3D();
    	
		@Override
		protected void map(LongWritable key, V value, Context context)
				throws IOException, InterruptedException {
			outBuf.setX(value.getX());
			outBuf.setY(value.getY());
			outBuf.setZ(value.getZ());
			context.write(key, outBuf);
		}
    }

    public static class RewriteParticleMapper<V extends IPhysicsParticle> extends Mapper<LongWritable,V,LongWritable,PhysicsParticle> {
    	private PhysicsParticle outBuf = new PhysicsParticle();
    	
		@Override
		protected void map(LongWritable key, V value, Context context)
				throws IOException, InterruptedException {
			outBuf.set(value);
			context.write(key, outBuf);
		}
    }
    
    public static class Point3DOutputFormat 
    extends WritableOutputFormat<LongWritable,Point3D> {}
    
    public static class PhysicsParticleOutputFormat 
    extends WritableOutputFormat<LongWritable,PhysicsParticle> {}    

    public int run(String[] args) throws Exception {
		boolean bad = false;
		int i = 0;
		String output = null;
		DataType dataType = null;
		boolean merge = false;
		boolean pointOutput = true;
		
		for ( ; i < args.length; ++i ) {
			if ( args[i].charAt(0) != '-' )
				break;
			
			if ( "-gas".equals(args[i]) ) {
				dataType = DataType.GAS;
			} else if ( "-dark".equals(args[i]) ) {
				dataType = DataType.DARK;
			} else if ( "-star".equals(args[i]) ) {
				dataType = DataType.STAR;
			} else if ( "-merge".equals(args[i]) ) {
				merge = true;
			} else if ( "-output".equals(args[i]) ) {
				output = args[++i];
			} else if ( "-point".equals(args[i]) ) {
				pointOutput = true;
			} else if ( "-particle".equals(args[i]) ) {
				pointOutput = false;
			} else {
				System.err.println("Unknown option: "+args[i]);
				bad = true;
			}
		}
		
		if ( bad || dataType == null ) {
			printUsage();
			System.exit(1);
		}
    	
		if ( output == null ) {
			output = String.format("%s-particle",dataType.name());
			System.err.println("WARNING: converted data will be written under "+output);
		}
		
        Job job = new Job(getConf(),"Rewrite"+dataType.name());

        job.setJarByClass(RewriteData.class);

        job.setInputFormatClass(dataType.getInputFormatClass());
        job.setOutputKeyClass(LongWritable.class);
        
        if ( pointOutput ) {
            job.setOutputFormatClass(Point3DOutputFormat.class);
            job.setOutputValueClass(Point3D.class);
            job.setMapperClass(RewritePointMapper.class);
        } else {
        	job.setOutputFormatClass(PhysicsParticleOutputFormat.class);
        	job.setOutputValueClass(PhysicsParticle.class);
        	job.setMapperClass(RewriteParticleMapper.class);
        }
        
        if ( merge ) {
        	job.setReducerClass(Reducer.class);
        	job.setNumReduceTasks(1);
        } else {
        	job.setNumReduceTasks(0);
        }

        for ( ; i < args.length; ++i ) {
            FileInputFormat.addInputPaths(job,args[i]);
        }
        FileOutputFormat.setOutputPath(job, new Path(output));

        job.submit();

        System.err.println("JOBID == "+ job.getJobID());

        return job.waitForCompletion(true) ? 0 : -1;
    }

    private void printUsage() {
    	System.out.printf("Usage: %s [-gas|-dark|-star] [-output outputDir] [-merge] input files ...\n",RewriteData.class);
	}

	public static void main(String[] args) throws Exception {
        int ret = ToolRunner.run(new RewriteData(),args);
        System.exit(ret);
    }
}
